package com.atguigu.gmall.realtime.tutorial.sql;

import com.atguigu.gmall.realtime.tutorial.beans.Sensor;
import com.atguigu.gmall.realtime.tutorial.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class SQLTEST_FileSystem {
    public static void main(String[] args)  throws Exception{
//TODO 0 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //CK设置

        //TODO 1 定义Table流环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,settings);

        tableEnv.executeSql("CREATE table sensor(id String,money BIGINT,temperature double) WITH ('connector' = 'filesystem','path'='K:\\Project\\flink_project\\flink_real_dataware\\gmall0821realtime\\src\\main\\java\\com\\atguigu\\gmall\\realtime\\tutorial\\sql\\sensor_filesystem.txt' ,'format' = 'csv')");




        TableResult result = tableEnv.sqlQuery("select id,money, temperature from sensor where temperature >35.0").execute();
        result.print();


        Table table = tableEnv.sqlQuery("select id,money, temperature from sensor");
        DataStream<Sensor> sensorReadingDataStream = tableEnv.toAppendStream(table, Sensor.class);
        sensorReadingDataStream.print("all >>>>>>>>");

        tableEnv.execute("xx");

    }
}
